package com.cloud.message;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * RocketMQ 事务消息生产端监听类
 *
 * @author: yzj
 **/
public abstract class AbstractMQTransactionProducer<T> extends AbstractMQConsumer<T> implements TransactionListener {

    private static Logger log = LoggerFactory.getLogger(AbstractMQTransactionProducer.class);

    /**
     * 执行本地事务，根据本地事务的执行结果，返回相应的LocalTransactionState状态
     *
     * @param message
     * @param data
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object data) {
        String msgKey = message.getKeys();
        log.info("开始执行本地事务{}", msgKey);
        LocalTransactionState state = getLocalTransaction(data);
        log.info("执行本地事务:{},执行状态:{},当前时间: {}", msgKey, state, System.currentTimeMillis());
        return state;
    }

    /**
     * 回查本地事务方法，适用于消息发送时返回UNKNOWN，事后补偿确认机制
     *
     * @param messageExt
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String msgKey = messageExt.getKeys();
        log.info("开始检查本地事务{}", msgKey);
        LocalTransactionState state = checkLocalTransaction(parseMessage(messageExt));
        log.info("执行支票交易:{},执行状态:{},当前时间:{}", msgKey, state, System.currentTimeMillis());
        return state;
    }

    /**
     * 执行本地事务hook方法，所有事务消息发送端都需要实现特定的事务内容
     *
     * @param data 发送消息对象
     * @return
     */
    public abstract LocalTransactionState getLocalTransaction(Object data);

    /**
     * 回查本地事务hook方法，所有事务消息发送端都需要实现，表示事务是否完成的标识
     *
     * @param t 发送消息对象
     * @return
     */
    public abstract LocalTransactionState checkLocalTransaction(T t);

}
